package com.huawei.bigdata.flink.examples.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Calendar;
import java.util.List;

public class CreateTable extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(CreateTable.class);
    private static org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
    public static Connection conn = null;
    private static Admin admin = null;
    public static Table table = null;
    public static void create(Configuration conf, String tableNamePrefix) throws IOException {
        Runnable runnable = new Runnable() {
            long sleep = 0L;
            @Override
            public void run() {
                while (true) {
                    try {
                        conn = ConnectionFactory.createConnection(conf);
                        long start = System.currentTimeMillis();
                        DateFormat df = new SimpleDateFormat("yyyy_MM");
                        Calendar ct = Calendar.getInstance();
                        ct.getTime();
                        String thisMonth = df.format(ct.getTime());
                        // ct.add(Calendar.MONTH, +1);
                        // String nextMonth = df.format(ct.getTime());
                        TableName thisMonthTableName = TableName.valueOf(tableNamePrefix+ thisMonth);
                        admin = conn.getAdmin();
                        table = conn.getTable(thisMonthTableName);
                        if (!admin.tableExists(thisMonthTableName)) {
                            createHBaseTable(admin, thisMonthTableName);
                        } else {
                            System.out.println("thisMonthTable already exists");
                        }
                        ct.add(Calendar.MONTH, 1);
                        ct.set(Calendar.DAY_OF_MONTH, 1);
                        ct.set(Calendar.HOUR_OF_DAY, 0);
                        ct.set(Calendar.MINUTE, 0);
                        ct.set(Calendar.SECOND, 0);
                        sleep = ct.getTime().getTime() - start;

                    } catch (IOException e) {
                        e.printStackTrace();
                    }finally {
                        if (table != null) {
                            try {
                                table.close();
                                System.out.println("HBase table  close!!!");
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if (admin != null) {
                            try {
                                admin.close();
                                System.out.println("HBase  admin close!!!");
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if (conn != null) {
                            try {
                                // Close the HBase connection.
                                conn.close();
                                System.out.println("HBase  conn close!!!");
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }

                    try {
                        Thread.sleep(sleep);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        Thread thread = new Thread(runnable);
        thread.start();
    }

    private static void createHBaseTable(Admin admin, TableName tableName) {
        System.out.println("Creating Table..." + tableName);
        HTableDescriptor tabDescriptor = new HTableDescriptor(tableName);
        HColumnDescriptor mob = new HColumnDescriptor("base");
        // Open mob function
//        mob.setMobEnabled(true);
//        // Set mob threshold
//        mob.setMobThreshold(10L);
        mob.setCompressionType(Compression.Algorithm.SNAPPY);
        mob.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);
        tabDescriptor.addFamily(mob);

        byte[][] splits = new byte[17][];
        String[] s = new String[]{"2", "4", "6", "8",  "a",  "c", "e",  "g",  "i", "k",  "m", "o", "q",  "s", "u", "w", "y"};
        List<String> index = Arrays.asList(s);
        for (int i = 0; i < splits.length; i++) {
            splits[i] = Bytes.toBytes(index.get(i));
        }
        try {
            admin.createTable(tabDescriptor, splits);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

